/////////////////////////////////////////////////////////////////////////
//
// Amuse Engine SDK - core/threads
// Copyright (c) 2014.  All Rights Reserved
//
// File:		AEThreadPool.cpp
// Author:		Gianluca Belardelli
// Date:		12/03/2014
//
/////////////////////////////////////////////////////////////////////////
#include <AEEngine.h>

AEThreadPool::AEThreadPool( void ) : m_lpWorkerThreads( 0 ) ,
									 m_lpWorkerThreadIds( 0 ) ,
									 m_uiNumThreads( 0 ) ,
									 m_lpWorkQueueSemaphore( 0 ),
									 m_lpThreadContexts( 0 ) ,
									 m_bInitialized( FALSE ) ,
									 m_uiNumActiveThreads( 0 )
{
}

AEThreadPool::~AEThreadPool( void )
{
	if( m_bInitialized )
		Shutdown();
}

void AEThreadPool::Initialize( AEUINT32 uiNumThreads )
{
	m_uiNumThreads = uiNumThreads;
	m_uiNumActiveThreads = uiNumThreads;

	if( m_uiNumThreads > 0 )
	{
		m_lpWorkQueueSemaphore = new AESemaphore( 0, 1024 );
		m_lpWorkerThreads = new AEThread*[ m_uiNumThreads ];
		m_lpWorkerThreadIds = new AEUINT32[ m_uiNumThreads ];
		m_lpWorkQueue = new AEWorkItemsQueue();

		m_lpThreadContexts = new AEThreadContext[ m_uiNumThreads ];

		for( AEUINT32 i=0; i < m_uiNumThreads; i++ )
		{
			m_lpThreadContexts[ i ].m_lpThreadPool = this;
	    
			// Inizializzazione eventi
			m_lpThreadContexts[ i ].m_lpStatusEvent = new AEEvent( TRUE );
			m_lpThreadContexts[ i ].m_lpSleepEvent = new AEEvent( TRUE );
			m_lpThreadContexts[ i ].m_lpWakeEvent = new AEEvent( TRUE );

			m_lpThreadContexts[ i ].m_lpWorkQueueSemaphore = m_lpWorkQueueSemaphore;
		}

		for( AEUINT32 i=0; i < m_uiNumThreads; i++ )
		{
			// NOTE: we should offset stack size of threads to avoid memory aliasing issues
			m_lpWorkerThreads[i] = new AEThread();
			m_lpWorkerThreadIds[ i ] = m_lpWorkerThreads[i]->StartThread( (AEThread::AEStartFunction *)this->workerThreadProc, ( void * )&m_lpThreadContexts[ i ], NULL );
		}
	}

	m_bInitialized = TRUE;
}

void AEThreadPool::Shutdown( void )
{
/*
	// wake up all of our threads so they can shut down
    SetActiveThreadCount( m_uiNumThreads );

    // send an event to shutdown the threads.
	AEHANDLE *lpThreadHandles = new AEHANDLE[m_uiNumThreads];

    for( AEUINT32 i = 0; i < m_uiNumThreads; i++ )
    {
        m_lpThreadContexts[ i ].m_csStatus = AEContextStatus::ExecShutdown;
        m_lpThreadContexts[ i ].m_lpStatusEvent->Set();
		lpThreadHandles[ i ] = m_lpWorkerThreads[ i ]->GetThreadHandle();
    }

    // wait for all the threads to complete

    WaitForMultipleObjects( m_uiNumThreads, lpThreadHandles, TRUE, INFINITE );
	delete [] lpThreadHandles;

	// clean queue
	while( !m_lpWorkQueue->IsEmpty() )
		PopAndProcessWorkItem( FALSE );

	if( m_lpWorkQueue != NULL )
	{
		delete m_lpWorkQueue;
		m_lpWorkQueue = NULL;
	}
    
    // clean up synch prims
    for( AEUINT32 i=0; i < m_uiNumThreads; i++ )
	{
        delete m_lpThreadContexts[ i ].m_lpStatusEvent;
		delete m_lpThreadContexts[ i ].m_lpSleepEvent;
		delete m_lpThreadContexts[ i ].m_lpWakeEvent;

		AEThread *lpThread = m_lpWorkerThreads[ i ];
		delete lpThread;
	}

	delete m_lpWorkQueue;
	delete m_lpWorkQueueSemaphore;

	delete [] m_lpThreadContexts;
	delete [] m_lpWorkerThreadIds;
	delete [] m_lpWorkerThreads;

    m_bInitialized = FALSE;
*/

}

void AEThreadPool::EnqueueWorkItem( AEThreadWorkItem *lpItem )
{
	AEASSERT( lpItem != NULL );

	{
		AEScopedLock Lock( m_csQueueSW );
		m_lpWorkQueue->Enqueue( lpItem );
	}

	if( m_lpWorkQueueSemaphore->Set() != AESemaphore::AESemaphore_Ok )
		AEASSERT( FALSE );
}

AEThreadWorkItem *AEThreadPool::RemoveWorkItem( void )
{
	AEThreadWorkItem *lpWorkItem = NULL;
	
	// create a block to scope a lock
	{
		AEScopedLock Lock( m_csQueueSW );
		
		// pop off the queue
		if( !m_lpWorkQueue->IsEmpty() )
		{
			m_lpWorkQueue->Dequeue(lpWorkItem);
			//AEASSERT( lpWorkItem != NULL );
			//m_lpWorkQueue->pop();
		}
	}
	
	return lpWorkItem;
}

void AEThreadPool::PrimaryThreadExecute( AEThreadContext::AEThreadCallback lpfnCallback, void *lpData )
{
/*	AEUINT32 uiOldThreadCount = m_uiNumActiveThreads;

	SetActiveThreadCount( m_uiNumThreads );

    m_uiCallbackCountdown = m_uiNumThreads;

    for( AEUINT32 i=0; i < m_uiNumThreads; i++ )
    {
        m_lpThreadContexts[ i ].m_lpfnThreadCallback = lpfnCallback;
        m_lpThreadContexts[ i ].m_lpThreadData = lpData;
        m_lpThreadContexts[ i ].m_lpCallbackCountdown = &m_uiCallbackCountdown;

        m_lpThreadContexts[ i ].m_csStatus = AEThreadContext::ExecCallback;
        //::SetEvent( m_lpThreadContexts[ i ].hStatusEvent );
		m_lpThreadContexts[ i ].m_lpStatusEvent->Set();
    }

	SetActiveThreadCount( uiOldThreadCount );

	while( m_uiCallbackCountdown > 0 )
        Sleep( 0 );
*/
}

void AEThreadPool::PopAndProcessWorkItem( AEBOOL32 bProcess )
{
/*	AEThreadWorkItem *lpWorkItem = RemoveWorkItem();
    if( lpWorkItem != NULL )
	{
		if( bProcess )
		{
			lpWorkItem->Process();
			lpWorkItem->SetStatus( AEThreadWorkItem::Completed );
		}

		if( lpWorkItem->IsDynamicallyAllocated() )
			delete lpWorkItem;
	}
*/
}
	
void AEThreadPool::SetActiveThreadCount( AEUINT32 uiCount )
{
	AEUINT32 uiNumThreads = uiCount;

	if( uiNumThreads > m_uiNumThreads )
	{
		uiNumThreads = m_uiNumThreads;
	}

	for( AEUINT32 i = 0; i < m_uiNumThreads; i++ )
    {
		if( ( i < uiNumThreads ) && ( i >= m_uiNumActiveThreads ) )
		{
			//---------------------------------------------------------------------
			// Set the thread's sleep event to signaled which wakes it up
			//
			//SetEvent( m_pThreadContext[ i ].hWakeEvent );
			m_lpThreadContexts[ i ].m_lpWakeEvent->Set();
		}

		if( ( i < m_uiNumActiveThreads ) && ( i >= uiNumThreads ) )
		{
			//---------------------------------------------------------------------
			// Set the thread's sleep event to nonsignaled which puts it to sleep
			//
			//SetEvent( m_pThreadContext[ i ].hSleepEvent );
			m_lpThreadContexts[ i ].m_lpSleepEvent->Set();
		}
	}

	m_uiNumActiveThreads = uiNumThreads;
}

AEUINT32 AESTDCALL AEThreadPool::workerThreadProc( void *lpContext ) 
{
 /*   AEHANDLE hSyncVector[3]; 
    AEUINT32 uiEventStatus = 0;
    AEThreadContext *lpThreadContext = ( AEThreadContext* )lpContext;
    AEThreadPool *lpThreadPool = lpThreadContext->m_lpThreadPool;

    hSyncVector[0] = lpThreadContext->m_lpWorkQueueSemaphore->GetHandle();
    hSyncVector[1] = lpThreadContext->m_lpStatusEvent->GetHandle();
	hSyncVector[2] = lpThreadContext->m_lpSleepEvent->GetHandle();

	AEUINT32 uiThreadId = GetCurrentThreadId();
	printf(" THREAD %d: Started\n", uiThreadId );

#pragma warning (push)
#pragma warning (disable: 4127)
    while( TRUE )
#pragma warning (pop)
    {
        uiEventStatus = WaitForMultipleObjects( 3, hSyncVector, FALSE, INFINITE ); 
        if( uiEventStatus == WAIT_TIMEOUT )
        {
            continue;
        }
        
        switch( uiEventStatus - WAIT_OBJECT_0 )
        { 
			case 0:
				// data available on the work queue
				printf(" THREAD %d: PopAndProcessWorkItem\n", uiThreadId );
				lpThreadPool->PopAndProcessWorkItem();
				break;

			case 1:
				// time to shutdown or do a callback
				if( lpThreadContext->m_csStatus == AEThreadContext::AEContextStatus::ExecShutdown )
				{
					printf(" THREAD %d: ExecShutdown\n", uiThreadId );
					return 0;
				}
				else if( lpThreadContext->m_csStatus == AEThreadContext::AEContextStatus::ExecCallback )
				{
					printf(" THREAD %d: ExecCallback\n", uiThreadId );
					lpThreadContext->m_lpfnThreadCallback( lpThreadContext->m_lpThreadData );
					::InterlockedDecrement( ( PLONG )lpThreadContext->m_lpCallbackCountdown );
					lpThreadContext->m_lpStatusEvent->Reset();
				}
				break;

			case 2:
				// time to sleep
				printf(" THREAD %d: Sleep\n", uiThreadId );
				lpThreadContext->m_lpSleepEvent->Reset();
				lpThreadContext->m_lpWakeEvent->Reset();
				//ResetEvent( pThreadContext->hSleepEvent );
				//ResetEvent( pThreadContext->hWakeEvent );
				lpThreadContext->m_lpWakeEvent->WaitInfinite();
				//WaitForSingleObject( pThreadContext->hWakeEvent, INFINITE );
				break;

			default:
				AEASSERT( FALSE ); // problem here
				return 0;
		}
	}
*/
}